#include "UbusEvent.h"

static struct blob_buf b;
static struct ubus_event_handler listener;
static struct uloop_timeout retry;

void ubus_probe_device_event(struct ubus_context *ctx, struct ubus_event_handler *ev, const char *type, struct blob_attr *msg)
{
    char *str;
    if (!msg)
        return;

    str = blobmsg_format_json(msg, true);

    printf("{ \"%s\": %s }\n", type, str);

    auto ps = UBusCallbackStaticManager::GetEventsMap().equal_range(type);
    for (auto it = ps.first; it != ps.second; it++)
    {
        it->second->onReceive(str);
    }

    free(str);
}

void ubus_reconn_timer(struct uloop_timeout *timeout) {}

void ubus_connection_lost(struct ubus_context *ctx)
{
    retry.cb = ubus_reconn_timer;
    if (0 != ubus_reconnect(ctx, nullptr))
    {
        uloop_timeout_set(&retry, 1000);
        return;
    }
    ubus_add_uloop(ctx);
}

UbusEvent::UbusEvent()
{
    ubus_init(nullptr);
    memset(&listener, 0, sizeof(listener));
    listener.cb = ubus_probe_device_event;
    threadRunning = false;
}
UbusEvent::~UbusEvent()
{
    ubus_done();
}

int UbusEvent::ubus_init(const char *path)
{
    uloop_init();
    sock_path = path;

    ctx = ubus_connect(path);
    if (!ctx)
    {
        printf("ubus connect failed\n");
        return -1;
    }

    printf("connected as %08x\n", ctx->local_id);
    ctx->connection_lost = ubus_connection_lost;

    return 0;
}

int UbusEvent::ubus_done()
{
    if (ctx)
        ubus_free(ctx);

    return 0;
}

int UbusEvent::ubus_register_events(const char *key)
{
    int ret = 0;

    return 0;
}

void UbusEvent::StartUloopThread()
{
    std::cout << "start thread" << std::endl;
    ubus_add_uloop(ctx);
#ifdef FD_CLOEXEC
    fcntl(ctx->sock.fd, F_SETFD, fcntl(ctx->sock.fd, F_GETFD) | FD_CLOEXEC);
#endif
    uloopThread = std::thread([&]() {
        threadRunning = true;
        uloop_run();
        uloop_done();
        threadRunning = false;
    });
    uloopThread.detach();
}

void UbusEvent::StopUloopThread()
{
    uloop_end();
    if (uloopThread.joinable())
    {
        uloopThread.join();
    }
}

int UbusEvent::SendMsg(const std::string &key, const std::string &json_msg)
{
    std::lock_guard<std::mutex> lock(m_mutex);

    blob_buf_init(&b, 0);
    blobmsg_add_json_from_string(&b, json_msg.c_str());
    return ubus_send_event(ctx, key.c_str(), b.head);
}

void UbusEvent::Subscribe(const std::string &key, IMessageReceiver *callback)
{
    if (key.empty())
        return;

    std::cout << key << " ... " << UBusCallbackStaticManager::GetEventsMap().count(key) << std::endl;

    if (!ctx)
        return;

    if (0 == UBusCallbackStaticManager::GetEventsMap().count(key))
    {
        if (0 != ubus_register_event_handler(ctx, &listener, key.c_str()))
            return;
        std::cout << "register event ok." << std::endl;
    }

    UBusCallbackStaticManager::addCallbackHandler(key, callback);

    if (!threadRunning)
    {
        StartUloopThread();
    }
}

void UbusEvent::unSubscribe(const std::string &key, IMessageReceiver *callback)
{
    UBusCallbackStaticManager::removeCallbackHandler(key, callback);
}
